package com.shujia.filter;

import com.shujia.been.Student;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class Code01ValueFilter {

    static Connection connection;

    static {
        // 创建连接需要给定连接信息，所以需要给定配置类
        Configuration conf = new Configuration();

        // 由于对HBASE进行操作时，需要连接其Zookeeper 那么需要传入Zookeeper的IP
        conf.set("hbase.zookeeper.quorum", "node1,node2,master");
        // 获取连接对象 通过ConnectionFactory来创建连接对象

        try {
            connection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static boolean tableExists(Admin admin, TableName tableName) {
        boolean exists = false;
        try {
            exists = admin.tableExists(tableName);
            if (exists) {
                System.out.println("表存在..");
            } else {
                System.out.println("表不存在");
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
        return exists;
    }




    // 需求：针对学生表中Value进行过滤，得到施姓的学生
    public static void main(String[] args) throws IOException {

        Admin admin = connection.getAdmin();
        TableName tableName = TableName.valueOf("api:tb1");

        if (tableExists(admin, tableName)) {
            Table table = connection.getTable(tableName);


            Scan scan = new Scan();


            /**
             * 创建ValueFilter时需要传入两个参数：
             *  public ValueFilter(final CompareOp valueCompareOp,
             *       final ByteArrayComparable valueComparator)
             * 第一个参数：
             * public enum CompareOp { LESS  LESS_OR_EQUAL EQUAL NOT_EQUAL ...}
             *
             * 第二参数
             *  BinaryComparator
             *  按字节索引顺序比较指定字节数组，采用Bytes.compareTo(byte[])
             *
             * BinaryPrefixComparator
             *    通BinaryComparator，只是比较左端前缀的数据是否相同
             *
             * NullComparator
             *判断给定的是否为空
             *
             * ###### BitComparator
             *
             * > 按位比较
             *
             * RegexStringComparator
             *    提供一个正则的比较器，仅支持 EQUAL 和 非EQUAL
             *
             * SubstringComparator
             *
             *  判断提供的子串是否出现在Value中
             *
             *
             */

            // 由于对字符串进行比较，可以使用正则表达式来比较
            // 注意：ValueFilter是拿所有列的Value进行比较
            //         并且在最终的结果中，只获取到对应列的数据
            ValueFilter valueFilter = new ValueFilter(
                    CompareFilter.CompareOp.EQUAL
                    , new RegexStringComparator("^施")
            );

            // 创建扫描器时，可以指定其过滤器
            scan.setFilter(valueFilter);

            ResultScanner scanner = table.getScanner(scan);

            printData(scanner);

            admin.close();
            table.close();
            connection.close();

        }


    }

    public static void printData(ResultScanner scanner){
        for (Result result : scanner) {
            String rowKey = Bytes.toString(result.getRow());

            String name = Bytes.toString(result.getValue("info1".getBytes(StandardCharsets.UTF_8)
                    , "name".getBytes(StandardCharsets.UTF_8)));

//            int age = Bytes.toInt(result.getValue("info1".getBytes(StandardCharsets.UTF_8)
//                    , "age".getBytes(StandardCharsets.UTF_8)));
//
//            String gender = Bytes.toString(result.getValue("info1".getBytes(StandardCharsets.UTF_8)
//                    , "gender".getBytes(StandardCharsets.UTF_8)));
//
//            String clazz = Bytes.toString(result.getValue("info2".getBytes(StandardCharsets.UTF_8)
//                    , "clazz".getBytes(StandardCharsets.UTF_8)));

            Student student = new Student(rowKey, name, 0, null, null);
            System.out.println(student);

//            System.out.println("name:"+name+" age:"+age+" gender:"+gender+" clazz:"+clazz);

        }
    }

}
